package com.atguigu.task;

import com.atguigu.userprofile.MysqlUtil.MyClickhouseUtil;
import com.atguigu.userprofile.MysqlUtil.MyPropsUtil;
import com.atguigu.userprofile.bean.Taginfo;
import com.atguigu.userprofile.constant.ConstCode;
import com.atguigu.userprofile.dao.TaginfoDao;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
/**
 * 提前在clickhouse中建表
 * */
public class Taskbitmap {
    public static void main(String[] args) {
        SparkConf taskBitmapApp = new SparkConf().setAppName("task_bitmap_app");
        SparkSession orCreate = SparkSession.builder().config(taskBitmapApp).enableHiveSupport().getOrCreate();
        //1.传入外部参数
        String taskid = args[0];
        String basidate = args[1];
        //2.明确标签列
        List<Taginfo> taginfos = TaginfoDao.selectTagInfosByTaskWithStatusEnable();


        String bigIntBitmapTable = "user_tag_value_bigint";
        String stringBitmapTable = "user_tag_value_string";
        String dateBitmapTable = "user_tag_value_date";
        String decimalBitmapTable = "user_tag_value_decimal";



        /**
         *         // 3. 创建bitmap表
         *         //  tag_code  tag_value  us  dt
         *         //  提前建表 , bitmap表的结构是固定的。
         *         //  留历史数据， 通过日期做一级分区， tag_code做二级分区.
         *         //  order by tag_value
         *
         *         // 还原标签值类型:  bigint  decimal(16,2) string date
         *         /*
         *            create table user_tag_value_string(
         *                 tag_code String,
         *                 tag_value String ,
         *                 us AggregateFunction(groupBitmap,UInt64),
         *                 dt String
         *             )engine=AggregatingMergeTree()
         *             partition by  (dt, tag_code)
         *             order by (tag_value);
         *
         *             create table  user_tag_value_bigint(
         *                 tag_code String,
         *                 tag_value UInt64 ,
         *                 us AggregateFunction(groupBitmap,UInt64),
         *                 dt String
         *             )engine=AggregatingMergeTree()
         *             partition by  (dt, tag_code)
         *             order by (tag_value);
         *
         *             create table  user_tag_value_decimal(
         *                 tag_code String,
         *                 tag_value Decimal64(2) ,
         *                 us AggregateFunction(groupBitmap,UInt64),
         *                 dt String
         *             )engine=AggregatingMergeTree()
         *             partition by  (dt, tag_code)
         *             order by (tag_value);
         *
         *             create table  user_tag_value_date(
         *                 tag_code String,
         *                 tag_value Date ,
         *                 us AggregateFunction(groupBitmap,UInt64),
         *                 dt String
         *             )engine=AggregatingMergeTree()
         *             partition by  (dt, tag_code)
         *             order by (tag_value);
         *          */
        //4.组织语句进行 insert.....select.......
        ArrayList<Taginfo> bigIntTaginfos = new ArrayList<>();
        ArrayList<Taginfo> stringTaginfos = new ArrayList<>();
        ArrayList<Taginfo> dateTaginfos = new ArrayList<>();
        ArrayList<Taginfo> decimalTaginfos = new ArrayList<>();

        for (Taginfo taginfo : taginfos) {
            String tagValueType = taginfo.getTagValueType();
            switch (tagValueType) {
                case ConstCode.TAG_VALUE_TYPE_BIGINT:
                    bigIntTaginfos.add(taginfo);
                    break;
                case ConstCode.TAG_VALUE_TYPE_DECIMAL:
                    decimalTaginfos.add(taginfo);
                    break;
                case ConstCode.TAG_VALUE_TYPE_STRING:
                    stringTaginfos.add(taginfo);
                       break;
                case ConstCode.TAG_VALUE_TYPE_DATE:
                    dateTaginfos.add(taginfo);
                    break;
            }
        }


        String bigIntInsertSelect  = tagInfosToInsertSelectSql(bigIntTaginfos, bigIntBitmapTable, basidate);
        String dateInsertSelect  = tagInfosToInsertSelectSql(dateTaginfos, dateBitmapTable, basidate);
        String decimalsertSelect  = tagInfosToInsertSelectSql(decimalTaginfos, decimalBitmapTable, basidate);
        String stringInsertSelect  = tagInfosToInsertSelectSql(stringTaginfos, stringBitmapTable, basidate);

        String bigintdeletaTable="alter table "+bigIntBitmapTable+" delete where dt= '"+basidate+"'";
        String stringdeletaTable="alter table "+stringBitmapTable+" delete where dt= '"+basidate+"'";
        String datedeletaTable="alter table "+dateBitmapTable+" delete where dt= '"+basidate+"'";
        String decimaldeletaTable="alter table "+decimalBitmapTable+" delete where dt= '"+basidate+"'";

        //执行sql
        executeSql(bigIntInsertSelect,bigintdeletaTable);
        executeSql(decimalsertSelect,decimaldeletaTable);
        executeSql(stringInsertSelect,stringdeletaTable);
        executeSql(dateInsertSelect,datedeletaTable);
    }
    public static void executeSql(String insertselect, String deletesql){
        MyClickhouseUtil.executesql(deletesql);
        if(insertselect !=null){
            MyClickhouseUtil.executesql(insertselect);
        }
    }


private static  String tagInfosToInsertSelectSql(List<Taginfo> tagInfos, String bitmapTable , String busiDate ){
        if(tagInfos.isEmpty()){
            return null;
        }
        String arrayJoinSql = null ;
        if("user_tag_value_bigint".equals(bitmapTable) || "user_tag_value_decimal".equals(bitmapTable)){
            arrayJoinSql = tagInfos.stream().map(
                    tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "',if( " + tagInfo.getTagCode().toLowerCase() + " == '' , '0'," + tagInfo.getTagCode().toLowerCase() + "))"
            ).collect(Collectors.joining(","));
        }else if("user_tag_value_date".equals(bitmapTable)){
            arrayJoinSql = tagInfos.stream().map(
                    tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "',if(" + tagInfo.getTagCode().toLowerCase() + " == '', '1970-01-01', " + tagInfo.getTagCode().toLowerCase()+ "))"
            ).collect(Collectors.joining(","));
        }else{
            arrayJoinSql = tagInfos.stream().map(
            tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "',if(" + tagInfo.getTagCode().toLowerCase() + " == '','无'," + tagInfo.getTagCode().toLowerCase() + "))").collect(Collectors.joining(","));
        }
         String ckdbname= MyPropsUtil.get(ConstCode.CK_DBNAME);
        String selectTable ="up_merge_" + busiDate.replace("-", "") ;
    // insert into user_tag_value_bitmap
    //     * select   t1.tt.1  as tag_code , t1.tt.2 as tag_value  , groupBitmapState(t1.uid) from
    //     *  (select uid , arrayJoin( [ ('gender' , gender) , ('agegroup' , agegroup ) , ('favor' , favor) ] ) as tt from user_tag_merge) t1
    //     * group by t1.tt.1 , t1.tt.2
        String insertSelectSql ="insert into  "+ckdbname+"."+bitmapTable+
                " select t1.tt.1 as tag_code , t1.tt.2 as tag_value , groupBitmapState(cast(t1.uid as UInt64)),'"+
                busiDate +"' from "+
                " (select uid ,arrayJoin(["+arrayJoinSql+" ]) as tt from "+selectTable+"" +
                " where uid != ''  )t1 "+
                " group by t1.tt.1 ,t1.tt.2";
        System.out.println("insertSelectSql :"+insertSelectSql );
            return insertSelectSql;
        }
}
